import com.zhenjun.domin.Item;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.ArrayList;
import java.util.Random;

/**
 * @author wangzj
 * @description 自定义实时流的数据源
 * @date 2020/7/10 0:27
 */
public class MyStreamingSource implements SourceFunction<Item> {
    private boolean isRunning = true;

    /**
     * 生成数据源
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void run(SourceContext<Item> ctx) throws Exception {
        while (isRunning) {
            Item item = generateItem();
            ctx.collect(item);
            //每秒产生一条数据
            Thread.sleep(1000);
        }
    }

    /**
     * 取消数据源
     */
    @Override
    public void cancel() {
        isRunning = false;
    }

    //随机产生一条商品数据
    private Item generateItem() {
        //随机生成100内的整数
        int i = new Random().nextInt(100);
//        Item item = new Item();
//        item.setName("name" + i);

        //Sreaming SQL统计数据
        ArrayList<String> list = new ArrayList();
        list.add("HAT");
        list.add("TIE");
        list.add("SHOE");
        Item item = new Item();
        item.setName(list.get(new Random().nextInt(3)));

        item.setId(i);
        return item;
    }

}

/**
 * 主函数执行
 */
class StreamingDome {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //获取数据源
        DataStreamSource<Item> text = env.addSource(new MyStreamingSource()).setParallelism(1);

        //1、将流式数据使用map转化
//        DataStream<Item> item = text
//                .map((MapFunction<Item,Item>) value -> value);

        //2、map的使用
//        SingleOutputStreamOperator<Object> item = text
//                //Item输入类型，Object输出类型
//                .map(new MapFunction<Item, Object>() {
//                    @Override
//                    public Object map(Item value) throws Exception {
//                        return value.getName();
//                    }
//                });

        //3、使用map的lambda表达式，编程更简洁
//        SingleOutputStreamOperator<String> item = text
//                .map(i -> i.getName());

        //4、自定义map，继承RichMapFunction
//        SingleOutputStreamOperator<String> item=text.map(new MyMapFunction());

        //5、FlaMap
//        SingleOutputStreamOperator<Object> item = text
//                .flatMap(new FlatMapFunction<Item, Object>() {
//                    @Override
//                    public void flatMap(Item value, Collector<Object> out) throws Exception {
//                        String name = value.getName();
//                        Map<String, Long> map = new HashMap<>();
//                        map.put(name, 1L);
//                        out.collect(map);
//                    }
//                });

        //6、Filter(返回id为偶数的数据)
//        SingleOutputStreamOperator<Item> item = text
//                .filter(value -> value.getId() % 2 == 0);

        //7、Reduce聚合函数
        SingleOutputStreamOperator<Item> item = text
                .map(value -> new Item(value.getName(), 1))
                //按照name聚合
                .keyBy(value -> value.getName())
                .reduce((value1, value2) -> new Item(value1.getName(), value1.getId() + value2.getId()));

        //打印结果
        item.print().setParallelism(1);
        env.execute("Streaming Item");
    }
}

/**
 * 自定义map
 */
class MyMapFunction extends RichMapFunction<Item, String> {
    @Override
    public String map(Item value) throws Exception {
        return value.getName();
    }
}